/* GStreamer
 * Copyright (C) 2022 FIXME <fixme@example.com>
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Library General Public
 * License as published by the Free Software Foundation; either
 * version 2 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Library General Public License for more details.
 *
 * You should have received a copy of the GNU Library General Public
 * License along with this library; if not, write to the
 * Free Software Foundation, Inc., 51 Franklin Street, Suite 500,
 * Boston, MA 02110-1335, USA.
 */
/**
 * SECTION:element-gstdsoftbussink
 *
 * The dsoftbussink element does FIXME stuff.
 *
 * <refsect2>
 * <title>Example launch line</title>
 * |[
 * gst-launch-1.0 -v fakesrc ! dsoftbussink ! FIXME ! fakesink
 * ]|
 * FIXME Describe what the pipeline does.
 * </refsect2>
 */

#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

#include <gst/gst.h>
#include <gst/base/gstbasesink.h>
#include "gstdsoftbussink.h"

GST_DEBUG_CATEGORY_STATIC (gst_dsoftbussink_debug_category);
#define GST_CAT_DEFAULT gst_dsoftbussink_debug_category

/* prototypes */


static void gst_dsoftbussink_set_property (GObject * object,
    guint property_id, const GValue * value, GParamSpec * pspec);
static void gst_dsoftbussink_get_property (GObject * object,
    guint property_id, GValue * value, GParamSpec * pspec);
static void gst_dsoftbussink_dispose (GObject * object);
static void gst_dsoftbussink_finalize (GObject * object);

static gboolean gst_dsoftbussink_propose_allocation (GstBaseSink * sink,
    GstQuery * query);
static gboolean gst_dsoftbussink_start (GstBaseSink * sink);
static gboolean gst_dsoftbussink_stop (GstBaseSink * sink);
static gboolean gst_dsoftbussink_unlock (GstBaseSink * sink);
static gboolean gst_dsoftbussink_unlock_stop (GstBaseSink * sink);
static gboolean gst_dsoftbussink_event (GstBaseSink * sink, GstEvent * event);
static GstFlowReturn gst_dsoftbussink_render (GstBaseSink * sink,
    GstBuffer * buffer);
static bool gst_dsoftbussink_detect(GstDsoftbussink * sink);
static gpointer gst_dsoftbussink_detect_thread(GstDsoftbussink *dsoftbussink);

enum
{
  PROP_0,
  PROP_PKGNAME,
  PROP_SESSIONNAME,
  PROP_PUBID,
  PROP_STARTVIDEO,
};

/* pad templates */

static GstStaticPadTemplate gst_dsoftbussink_sink_template =
GST_STATIC_PAD_TEMPLATE ("sink",
    GST_PAD_SINK,
    GST_PAD_ALWAYS,
    GST_STATIC_CAPS_ANY
    );
#define SESSION_MAX 65536
#define SINK_MAX 256
static GstDsoftbussink *g_sessionArray[SESSION_MAX] = {NULL};
static int g_sink_index = 0;
static GstDsoftbussink  *g_sinkArray[SINK_MAX] = {NULL};

/* class initialization */

G_DEFINE_TYPE_WITH_CODE (GstDsoftbussink, gst_dsoftbussink, GST_TYPE_BASE_SINK,
  GST_DEBUG_CATEGORY_INIT (gst_dsoftbussink_debug_category, "dsoftbussink", 0,
  "debug category for dsoftbussink element"));

static void
gst_dsoftbussink_class_init (GstDsoftbussinkClass * klass)
{
  GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
  GstBaseSinkClass *base_sink_class = GST_BASE_SINK_CLASS (klass);

  /* Setting up pads and setting metadata should be moved to
     base_class_init if you intend to subclass this class. */
  gst_element_class_add_static_pad_template (GST_ELEMENT_CLASS(klass),
      &gst_dsoftbussink_sink_template);

  gst_element_class_set_static_metadata (GST_ELEMENT_CLASS(klass),
      "FIXME Long name", "Generic", "FIXME Description",
      "FIXME <fixme@example.com>");

  gobject_class->set_property = gst_dsoftbussink_set_property;
  gobject_class->get_property = gst_dsoftbussink_get_property;
  gobject_class->dispose = gst_dsoftbussink_dispose;
  gobject_class->finalize = gst_dsoftbussink_finalize;
  base_sink_class->propose_allocation = GST_DEBUG_FUNCPTR (gst_dsoftbussink_propose_allocation);
  base_sink_class->start = GST_DEBUG_FUNCPTR (gst_dsoftbussink_start);
  base_sink_class->stop = GST_DEBUG_FUNCPTR (gst_dsoftbussink_stop);
  base_sink_class->unlock = GST_DEBUG_FUNCPTR (gst_dsoftbussink_unlock);
  base_sink_class->unlock_stop = GST_DEBUG_FUNCPTR (gst_dsoftbussink_unlock_stop);
  base_sink_class->event = GST_DEBUG_FUNCPTR (gst_dsoftbussink_event);
  base_sink_class->render = GST_DEBUG_FUNCPTR (gst_dsoftbussink_render);

  g_object_class_install_property (gobject_class, PROP_SESSIONNAME,
      g_param_spec_string ("session_name", "session_name", "dsoftbus session name",
          "gstreamer", G_PARAM_READWRITE));
  g_object_class_install_property (gobject_class, PROP_PKGNAME,
      g_param_spec_string ("pkg_name", "pkg_name", "dsoftbus pkg name",
          "gstreamer", G_PARAM_READWRITE));
  g_object_class_install_property (gobject_class, PROP_PUBID,
      g_param_spec_int ("publish_id", "publish_id", "dsoftbus publish id",
          1, 65535, 123, G_PARAM_READWRITE));
  g_object_class_install_property (gobject_class, PROP_STARTVIDEO,
      g_param_spec_boolean ("start_video", "start_video", "auto start video",
          TRUE, G_PARAM_READWRITE));
}

static void
gst_dsoftbussink_init (GstDsoftbussink *dsoftbussink)
{
  dsoftbussink->pkg_name = "gstreamer";
  dsoftbussink->session_name = "gstreamer";
  dsoftbussink->publish_id = 123;
  dsoftbussink->sessionId = -1;
  dsoftbussink->detect_thread_running = TRUE;
  g_sinkArray[g_sink_index] = dsoftbussink;
  g_sink_index ++;

  dsoftbussink->detect_queue = g_async_queue_new();
}

void
gst_dsoftbussink_set_property (GObject * object, guint property_id,
    const GValue * value, GParamSpec * pspec)
{
  GstDsoftbussink *dsoftbussink = GST_DSOFTBUSSINK (object);

  GST_DEBUG_OBJECT (dsoftbussink, "set_property");

  switch (property_id) {
    case PROP_SESSIONNAME:
      dsoftbussink->session_name = g_value_dup_string(value);
      break;
    case PROP_PKGNAME:
      dsoftbussink->pkg_name = g_value_dup_string(value);
      break;
    case PROP_PUBID:
      dsoftbussink->publish_id = g_value_get_int(value);
      break;
    case PROP_STARTVIDEO:
      dsoftbussink->enable_video = g_value_get_boolean(value);
      break;
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
      break;
  }
}

void
gst_dsoftbussink_get_property (GObject * object, guint property_id,
    GValue * value, GParamSpec * pspec)
{
  GstDsoftbussink *dsoftbussink = GST_DSOFTBUSSINK (object);

  GST_DEBUG_OBJECT (dsoftbussink, "get_property");

  switch (property_id) {
    case PROP_SESSIONNAME:
      g_value_set_string(value, dsoftbussink->session_name);
      break;
    case PROP_PKGNAME:
      g_value_set_string(value, dsoftbussink->pkg_name);
      break;
    case PROP_PUBID:
      g_value_set_int(value, dsoftbussink->publish_id);
      break;
    case PROP_STARTVIDEO:
      g_value_set_boolean(value, dsoftbussink->enable_video);
      break;
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
      break;
  }
}

void
gst_dsoftbussink_dispose (GObject * object)
{
  GstDsoftbussink *dsoftbussink = GST_DSOFTBUSSINK (object);

  GST_DEBUG_OBJECT (dsoftbussink, "dispose");

  /* clean up as possible.  may be called multiple times */

  G_OBJECT_CLASS (gst_dsoftbussink_parent_class)->dispose (object);
}

void
gst_dsoftbussink_finalize (GObject * object)
{
  GstDsoftbussink *dsoftbussink = GST_DSOFTBUSSINK (object);

  GST_DEBUG_OBJECT (dsoftbussink, "finalize");
  g_async_queue_unref(dsoftbussink->detect_queue);
  /* clean up object here */
  G_OBJECT_CLASS (gst_dsoftbussink_parent_class)->finalize (object);
}

/* propose allocation parameters for upstream */
static gboolean
gst_dsoftbussink_propose_allocation (GstBaseSink * sink, GstQuery * query)
{
  GstDsoftbussink *dsoftbussink = GST_DSOFTBUSSINK (sink);

  GST_DEBUG_OBJECT (dsoftbussink, "propose_allocation");

  return TRUE;
}

static int SessionOpened(int sessionId, int result)
{
	printf("<SessionOpened>CB: session %d open ret=%d\n", sessionId, result);
  if (sessionId >= SESSION_MAX)
  {
    printf("<SessionClosed>CB: session %d too large\n", sessionId);
    return 0;
  }
  if (result < 0)
  {
    return 0;
  }
  if (g_sessionArray[sessionId] == NULL)
  {
    char peerSessionName[50];
    GetPeerSessionName(sessionId, peerSessionName, 50);
    for (int i = 0; i < g_sink_index; i ++)
    {
      if (g_sinkArray[i]->sessionId < 0 && strcmp(peerSessionName, g_sinkArray[i]->session_name) == 0)
      {
        printf("store session by sessionId\r\n");
        g_sessionArray[sessionId] = g_sinkArray[i];
        g_sessionArray[sessionId]->sessionId = sessionId;
      }
      else
      {
        printf("per session name %s not match need session name %s\r\n", peerSessionName, g_sinkArray[i]->session_name);
      }
    }
  }
  else
  {
    printf("store session by ID\r\n");
    g_sessionArray[sessionId]->sessionId = sessionId;
  }
  printf("<SessionClosed>CB: session %d match package %s\n",
        sessionId, g_sessionArray[sessionId]->pkg_name);
	return 0;
}

static void SessionClosed(int sessionId)
{
	printf("<SessionClosed>CB: session %d closed\n", sessionId);
  if (g_sessionArray[sessionId] != NULL)
  {
    g_sessionArray[sessionId]->sessionId = -1;
  }
  else
  {
    printf("<SessionClosed>CB: session %d not registed\n", sessionId);
  }
}

static void ByteRecived(int sessionId, const void *data, unsigned int dataLen)
{
	printf("<ByteRecived>CB: session %d received %u bytes data=%s\n", sessionId, dataLen, (const char *)data);
  if (strstr(data, "land") > 0)
  {
    send_signal(CMD_LAND);
  }
  else if (strstr(data, "takeoff") > 0)
  {
    send_signal(CMD_TAKEOFF);
  }
  else if (strstr(data, "arm") > 0)
  {
    send_signal(CMD_ARM);
  }
  else if (strstr(data, "startVideo") > 0)
  {
    g_sessionArray[sessionId]->enable_video = TRUE;
  }
  else if (strstr(data, "endVideo") > 0)
  {
    g_sessionArray[sessionId]->enable_video = FALSE;
  }
}
static void MessageReceived(int sessionId, const void *data, unsigned int dataLen)
{
	printf("<MessageReceived>CB: session %d received %u bytes message=%s\n", sessionId, dataLen, (const char *)data);
}
static const ISessionListener sessionCB = {
	.OnSessionOpened = SessionOpened,
	.OnSessionClosed = SessionClosed,
	.OnBytesReceived = ByteRecived,
	.OnMessageReceived = MessageReceived,
};


static bool
gst_dsoftbussink_detect(GstDsoftbussink * sink)
{
  NodeBasicInfo **dev =  &sink->dev;
  int32_t *num = &sink->dev_num;
  GetAllNodeDeviceInfo(sink->pkg_name, dev, num);
  printf("<GetAllNodeDeviceInfo>return %d Node\n", *num);
	for (int i = 0; i < *num; i++) {
    // char devId[UDID_BUF_LEN];
    printf("<num %d>deviceName=%s\n", i + 1, sink->dev[i].deviceName);
		printf("\tnetworkId=%s\n", sink->dev[i].networkId);
    // printf("%d,\r\n", i + 1);
    // if (GetNodeKeyInfo(sink->pkg_name, sink->dev[i].networkId, NODE_KEY_UDID, devId, UDID_BUF_LEN) == 0) {
		// 	printf("\tdevId=%s\n", devId);
		// }
		// printf("\tType=%d\n", sink->dev[i].deviceTypeId);
	}

  if (sink->dev_num) {
    int sessionId = -1;
    for (int i = 0; i < sink->dev_num; i ++)
    {
      sessionId = OpenSessionInterface("gstreamer", sink->session_name, sink->dev[i].networkId);
      if (sessionId >= 0 && sessionId < SESSION_MAX)
      {
        g_sessionArray[sessionId] = sink;
      }
      else
      {
        GST_ERROR_OBJECT(sink, "sessionId =%d mismatch", sessionId);
      }
    }

    return TRUE;
  }
  else {
    GST_INFO_OBJECT(sink, "GetAllNodeDeviceInfoInterface dev_num=%d\n", sink->dev_num);
    return FALSE;
  }
}

static void DeviceFound(const DeviceInfo *device)
{
	int i;
	printf("<DeviceFound>CB: Device has found\n");
	printf("\tdevId=%s\n", device->devId);
	printf("\tdevName=%s\n", device->devName);
	printf("\tdevType=%d\n", device->devType);
	printf("\taddrNum=%d\n", device->addrNum);
	for (i = 0; i < device->addrNum; i++) {
		printf("\t\taddr%d:type=%d,", i + 1, device->addr[i].type);
		switch (device->addr[i].type) { 
		case CONNECTION_ADDR_WLAN:
		case CONNECTION_ADDR_ETH:
			printf("ip=%s,port=%d,", device->addr[i].info.ip.ip, device->addr[i].info.ip.port);
			break;
		default:
			break;
		}
		printf("peerUid=%s\n", device->addr[i].peerUid);
	}
	printf("\tcapabilityBitmapNum=%d\n", device->capabilityBitmapNum);
	for (i = 0; i < device->addrNum; i++) {
		printf("\t\tcapabilityBitmap[%d]=0x%x\n", i + 1, device->capabilityBitmap[i]);
	}
	printf("\tcustData=%s\n", device->custData);

  // printf("g_sink_index=%d\n", g_sink_index);
  // for (int i = 0; i < g_sink_index; i ++)
  // {
  //   GstDsoftbussink *sink = g_sinkArray[i];
  //   if (sink->sessionId < 0)
  //   {
  //     printf("gst_dsoftbussink_detect%d\n", i);
  //     g_async_queue_push(sink->detect_queue, sink);
  //   }
  // }
}

static gpointer
gst_dsoftbussink_detect_thread(GstDsoftbussink *dsoftbussink)
{
  while (dsoftbussink->detect_thread_running)
  {
    if (dsoftbussink->sessionId < 0)
    {
      gst_dsoftbussink_detect(dsoftbussink);
    }
    g_usleep(1000*1000);
    // g_async_queue_pop (dsoftbussink->detect_queue);
    // GST_ERROR_OBJECT(dsoftbussink, "detect thread one loop");
    // if (dsoftbussink->sessionId < 0)
    // {
    //   GST_ERROR_OBJECT(dsoftbussink, "try to attach new session");
    //   gst_dsoftbussink_detect(dsoftbussink);
    // }
  }
  return NULL;
}
/* start and stop processing, ideal for opening/closing the resource */
static gboolean
gst_dsoftbussink_start (GstBaseSink * sink)
{
  int ret;
  GstDsoftbussink *dsoftbussink = GST_DSOFTBUSSINK (sink);

  GST_DEBUG_OBJECT (dsoftbussink, "start");
  SubscribeInfo info = {
		.subscribeId = dsoftbussink->publish_id,
		.mode = DISCOVER_MODE_ACTIVE,
		.medium = COAP,
		.freq = LOW,
		.isSameAccount = false,
		.isWakeRemote = false,
		.capability = "osdCapability",
		.capabilityData = (unsigned char*)cData,
		.dataLen = strlen(cData) + 1,
	};

	IDiscoveryCallback cb = {
		.OnDeviceFound = DeviceFound,
		.OnDiscoverFailed = DiscoveryFailed,
		.OnDiscoverySuccess = DiscoverySuccess,
	};


  
  ret = StartDiscovery(dsoftbussink->pkg_name, &info, &cb);
	if (ret) {
		GST_ERROR_OBJECT(sink, "DiscoveryInterface fail, ret=%d\n", ret);
		goto err_DiscoveryInterface;
	}

  // info.subscribeId = dsoftbussink->publish_id + 1;
  // info.mode = DISCOVER_MODE_PASSIVE;

  // ret = StartDiscovery(dsoftbussink->pkg_name, &info, &cb);
	// if (ret) {
	// 	GST_ERROR_OBJECT(sink, "DiscoveryInterface fail, ret=%d\n", ret);
	// 	goto err_DiscoveryInterface;
	// }

  ret = CreateSessionServer(dsoftbussink->pkg_name, "gstreamer", &sessionCB);
	if (ret) {
		GST_ERROR_OBJECT(sink, "CreateSessionServer fail, ret=%d\n", ret);
		goto err_CreateSessionServer;
	}
  
	dsoftbussink->detect_thread = g_thread_new("detect thread", gst_dsoftbussink_detect_thread, dsoftbussink);
  // return gst_dsoftbussink_detect(dsoftbussink);
  return TRUE;
err_CreateSessionServer:
	StopDiscovery(dsoftbussink->pkg_name, dsoftbussink->publish_id);
  // StopDiscovery(dsoftbussink->pkg_name, dsoftbussink->publish_id + 1);
err_DiscoveryInterface:
	return FALSE;
}

static gboolean
gst_dsoftbussink_stop (GstBaseSink * sink)
{
  GstDsoftbussink *dsoftbussink = GST_DSOFTBUSSINK (sink);

  dsoftbussink->detect_thread_running = FALSE;
  g_async_queue_push(dsoftbussink->detect_queue, dsoftbussink);
  g_thread_join(dsoftbussink->detect_thread);
  GST_DEBUG_OBJECT (dsoftbussink, "stop");
  // for (int i = 0;i < g_sink_index; i ++)
  // {
  //   if (g_sinkArray[i] == dsoftbussink)
  //   {
  //     CloseSession(i);
  //   }
  // }
  CloseSession(dsoftbussink->sessionId);
  FreeNodeInfo(dsoftbussink->dev);
  RemoveSessionServer(dsoftbussink->pkg_name, "gstreamer");
  g_print("pkgname = %s, publishid = %d\r\n", dsoftbussink->pkg_name, dsoftbussink->publish_id);
  StopDiscovery(dsoftbussink->pkg_name, dsoftbussink->publish_id);
  // StopDiscovery(dsoftbussink->pkg_name, dsoftbussink->publish_id + 1);
  return TRUE;
}

/* unlock any pending access to the resource. subclasses should unlock
 * any function ASAP. */
static gboolean
gst_dsoftbussink_unlock (GstBaseSink * sink)
{
  GstDsoftbussink *dsoftbussink = GST_DSOFTBUSSINK (sink);

  GST_DEBUG_OBJECT (dsoftbussink, "unlock");

  return TRUE;
}

/* Clear a previously indicated unlock request not that unlocking is
 * complete. Sub-classes should clear any command queue or indicator they
 * set during unlock */
static gboolean
gst_dsoftbussink_unlock_stop (GstBaseSink * sink)
{
  GstDsoftbussink *dsoftbussink = GST_DSOFTBUSSINK (sink);

  GST_DEBUG_OBJECT (dsoftbussink, "unlock_stop");

  return TRUE;
}

/* notify subclass of event */
static gboolean
gst_dsoftbussink_event (GstBaseSink * sink, GstEvent * event)
{
  GstDsoftbussink *dsoftbussink = GST_DSOFTBUSSINK (sink);

  GST_DEBUG_OBJECT (dsoftbussink, "event");

  return TRUE;
}

static GstFlowReturn
gst_dsoftbussink_render (GstBaseSink * sink, GstBuffer * buffer)
{
  int ret;
  GstDsoftbussink *dsoftbussink = GST_DSOFTBUSSINK (sink);

  GST_DEBUG_OBJECT (dsoftbussink, "render");

  if (dsoftbussink->sessionId < 0)
  {
    GST_DEBUG_OBJECT (dsoftbussink, "session not ready");
    return GST_FLOW_OK;
  }
  if (dsoftbussink->enable_video == FALSE)
  {
    GST_DEBUG_OBJECT (dsoftbussink, "skip send video");
    return GST_FLOW_OK;
  }
  GstMapInfo *info = malloc(sizeof(GstMapInfo));
  memset(info, 0, sizeof(GstMapInfo));
  gst_buffer_map(buffer, info, GST_MAP_READ);
  GST_INFO_OBJECT(dsoftbussink, "SendBytes size=%d\n", info->size);
  ret = SendBytes(dsoftbussink->sessionId,  info->data, info->size);
  if (ret < 0)
  {
     GST_ERROR_OBJECT(dsoftbussink, "send fail, redetect next, ret = %d\n", ret);
  }
  gst_buffer_unmap(buffer, info);
  free(info);
  return GST_FLOW_OK;
}

static gboolean
plugin_init (GstPlugin * plugin)
{

  /* FIXME Remember to set the rank if it's an element that is meant
     to be autoplugged by decodebin. */
  return gst_element_register (plugin, "dsoftbussink", GST_RANK_NONE,
      GST_TYPE_DSOFTBUSSINK);
}

/* FIXME: these are normally defined by the GStreamer build system.
   If you are creating an element to be included in gst-plugins-*,
   remove these, as they're always defined.  Otherwise, edit as
   appropriate for your external plugin package. */
#ifndef VERSION
#define VERSION "0.0.FIXME"
#endif
#ifndef PACKAGE
#define PACKAGE "FIXME_package"
#endif
#ifndef PACKAGE_NAME
#define PACKAGE_NAME "FIXME_package_name"
#endif
#ifndef GST_PACKAGE_ORIGIN
#define GST_PACKAGE_ORIGIN "http://FIXME.org/"
#endif

GST_PLUGIN_DEFINE (GST_VERSION_MAJOR,
    GST_VERSION_MINOR,
    dsoftbussink,
    "FIXME plugin description",
    plugin_init, VERSION, "LGPL", PACKAGE_NAME, GST_PACKAGE_ORIGIN)

